/*  
  -- The Lime II Project -- 

  A tuplespaces-based middleware for coordinating agents and mobile hosts.
  Copyright (C) 2005.
  Gian Pietro Picco, Amy L. Murphy, Lorenzo Bellini.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
*/

package lime2.ca.localtest;
import lime2.ca.*;
import lime2.*;

import java.io.*;
import java.net.*;
import java.util.HashSet;

/**
 * A testing-oriented communication adapter based on plain UDP unicast; it simulates multicasting for an embedded list
 * of local ports. UDP is faster than TCP but as much reliable only in the local case. 
 * <br>
 * Please note that the unicast port is a UDP port; the multicast port is not opened.
 * 
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini 
 */

public class LocalTestCA implements ICommunicationAdapter
{	
 // "fake" multicast group	
 final static int[] mgroup =  new int[] { 1973, 2073, 2173, 2273, 2373, 2473, 2573, 2673, 2773, 2873 };

 // datagram socket
 static DatagramSocket sock;
 
 // loopback address
 static InetAddress localhost; 
 
 // internal queue for incoming remote ops
 static Queue queue; 
 
 // receiver
 static LocalTestReceiver ltr;
 
 // detector
 static LocalTestDetector ltd;
 
 /**
  * This method must be called before the communication adapter is used; it performs all the necessary initializations
  * (i.e. starting thread/s, establishing permanent connections, etc..) 
  * 
  * @throws LimeCommunicationException
  */
 public void init() throws LimeCommunicationException
 {		
  // sets port
  int port = LimeServer.getServer().getUnicastPort();
  
  // creates datagram socket
  sock = null;
  try { sock = new DatagramSocket(port); } 
  catch (IOException ioe) { throw new LimeCommunicationException("can't listen on port " + port + " (UDP)", ioe, Failures.CRITICAL); }
 	
  // checks port (must belong to the embedded list)  
  boolean found = false;
  for (int i = 0; !found && i < mgroup.length; i++) if (mgroup[i] == port) found = true;
  if (!found) throw new LimeError("invalid port (see embedded list in class LocalTestCa)"); 
  
  // initializations
  localhost = Location.HERE.getID().getHost();
  queue = new Queue();
  
  // threading 
  ltr = new LocalTestReceiver(sock);
  ltr.start();
  
  // creates detector
  ltd = new LocalTestDetector();
 }
		
 /**
  * Enables this host to join the system. The serializable object may serve to provide startup information for the 
  * joining process (i.e. a host to attach to in a dispatching tree). Please note that communication exceptions are 
  * not thrown; when this method is invoked, it is guaranteed that the LimeServer will join the system as soon as 
  * possible.
  * 
  * @param info startup information for the joining process
  */	
 public void joinSystem(Serializable info)
 {
  ltd.enable();
 }

 /** 
  * Enables this host to leave the system. Please note that communication exceptions are not thrown; when this method
  * is invoked, it is guaranteed that the LimeServer will eventually leave the system.
  */
 public void leaveSystem() 
 {
  ltd.disable();
 }

 /**
  * Declares that from now on this node is interested in everything that concerns a specific tuplespace. Please note 
  * that communication exceptions are not thrown; when this method is invoked, it is guaranteed that the LimeServer 
  * will join the group as soon as possible.
  * 
  * @param name name of the tuplespace we are interested in
  */ 
 public void joinGroup(String name) 
 {
  // does nothing	
 }
	 
 /**
  * Declares that from now on this node is uninterested in everything that concerns a specific tuplespace. Please note
  * that communication exceptions are not thrown; when this method is invoked, it is guaranteed that the LimeServer 
  * will eventually leave the group.
  * 
  * @param name name of the tuplespace we are no more interested in
  */ 
 public void leaveGroup(String name) 
 {
  // does nothing
 }

 /**
  * Sends an operation request to all the group members.
  * 
  * @param name name of the destination tuplespace 
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendOp(String name, RemoteOp d) throws LimeCommunicationException
 {
  sendGroup(d); 
 }

 /**
  * Sends an operation request to a specific host.
  * 
  * @param host destination host
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendOp(LimeServerID host, RemoteOp d) throws LimeCommunicationException
 {
  send(host, d); 
 }

 /**
  * Sends a reaction registration request to all group members.
  * 
  * @param name name of the destination tuplespace 
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendReaction(String name, RemoteOp d) throws LimeCommunicationException
 {
  sendGroup(d); 
 }

 /**
  * Sends a reaction registration request to a specific host.
  * 
  * @param host destination host
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendReaction(LimeServerID host, RemoteOp d) throws LimeCommunicationException
 {
  send(host, d); 
 }
 
 /**
  * Allows to bind a message to a previous operation request (i.e, to gather results).
  * 
  * @param host destination host
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendReply(LimeServerID host, RemoteOp d) throws LimeCommunicationException
 {
  send(host, d); 
 }
 
 /**
  * Returns a received RemoteOp (FIFO order, blocking).
  * 
  * @return RemoteOp
  */
 public RemoteOp getNextOp()
 {
  return (RemoteOp) queue.remove();	
 }
 
 /**
  * Deactivates the communication layer.
  */
 public void shutdown()
 {
  ltr.shutdown();	
 }
 
 /**
  * Returns whether an host is engaged.
  * 
  * @param host
  * @return <code>true</code> or <code>false</code>
  */
 public boolean queryHost(LimeServerID host)
 {
  return ltd.queryHost(host);
 }
 
 /**
  * Adds a new (discovered) host.
  * 
  * @param host
  */
 public void addHost(LimeServerID host)
 {
  ltd.addHost(host);
 }
  
 /**
  * Removes a (probable) dead host.
  * 
  * @param host
  */
 public void removeHost(LimeServerID host)
 {
  ltd.removeHost(host);
 }
 
 // unicast send; we rely on internal synchronization of datagram sockets 
 private void send(LimeServerID target, RemoteOp d) throws LimeCommunicationException
 {
  //  total failure ?	
  if (sock == null) throw new LimeCommunicationException(Failures.TOTAL);		 
 	
  // temporary variables
  byte[] data = Serializer.serialize(d);
  boolean end = false;
  int retries = 0;
  
  // partial failure ?
  while (!end)
  {	
   try 
   { 
   	sock.send(new DatagramPacket(data, data.length, localhost, target.getPort())); 
    end = true;
   }
   catch (NullPointerException npe) { throw new LimeCommunicationException(npe, Failures.TOTAL); } // socket closed !
   catch (IOException ioe) 
   { 
   	if (++retries >= LimeConstants.RETRIES) throw new LimeCommunicationException(ioe, Failures.PARTIAL); 
   }
  }
 }
  
 // multicast send (emulated); we rely on internal synchronization of datagram sockets 
 private void sendGroup(RemoteOp d) throws LimeCommunicationException
 {    
  // total failure ?	
  if (sock == null) throw new LimeCommunicationException(Failures.TOTAL);			
 	
  // temporary variables
  byte[] data = Serializer.serialize(d); 
    
  // partial failure ? (if one "host" cannot be reached, the group send is compromised)  
  for (int i = 0; i < mgroup.length; i++) 
  {
   // ... we reset "retries" for each target	
   boolean end = false;		
   int retries = 0;
   
   while (!end) 
   { 
    try 
    { 
   	 sock.send(new DatagramPacket(data, data.length, localhost, mgroup[i])); 
     end = true;
    }
    catch (NullPointerException npe) { throw new LimeCommunicationException(npe, Failures.TOTAL); } // socket closed ! 
    catch (IOException ioe) 
    { 
   	 if (++retries >= LimeConstants.RETRIES) throw new LimeCommunicationException(ioe, Failures.PARTIAL); 
    }
   }
  } 
 }

@Override
public HashSet<LimeServerID> multicast(LimeServerID[] hosts, RemoteOp d)
		throws LimeCommunicationException {
	return null;
	
}

@Override
public HashSet<LimeServerID> multicast(RemoteOp d) throws LimeCommunicationException {
	return null;
}
 
}